package com.atguigu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderDetail;
import com.atguigu.realtime.bean.OrderInfo;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.DimQuery;
import com.atguigu.realtime.util.MyJDBCUtil;
import com.atguigu.realtime.util.MyJedisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Map;

/**
 * @ClassName: DWMOrderWideApp
 * @Description:  V2.0     将数据缓存在redis中
 * @Author: kele
 * @Date: 2021/4/20 14:53
 *
 * 升级说明：
 *      针对连接phoenix时，需要频繁的访问hbase，每读一次需要连接phoenix，网络连接多，读性能变差
 *
 *      方案1：将数据缓存到本地状态
 *        实现思路：每次需要读取数据时，先从本地读取，本地没有，则通过phoenix读取，再将数据缓存到本地
 *
 *          优点：从本地读取，效率高
 *          缺点；1、随着数据增多，本地状态需要的内存更多，
 *               2、如果phoenix中表的数据发生变化（update）,在状态中无法感知    --不可避免，不能使用（理解：状态不能跨两个程序使用）
 *
 *      方案2：将数据缓存到redis中
 *          实现思路：先从phoenix中读取数据，读一条数据，就将数据添加到redis中，并设置过期时间，
 *                  当维度数据发生变化时，通过DWDdbAPP中程序读取，如果发现update操作，删除redis中相应的缓存的数据
 *
 *          优点：减少了hbase的压力
 *               可以在读取端检测数据变化并更新
 *
 *               综上所述：采用方案2
 **/
public class DWMOrderWideAppV2 extends BaseAppV2 {

    public static void main(String[] args) throws SQLException, ClassNotFoundException {

        new DWMOrderWideAppV2().init(30004,
                2,
                "DWMOrderWideApp",
                "DWMOrderWideApp",
                "dwd_order_info",
                "dwd_order_detail");
    }

    @Override
    protected void run(StreamExecutionEnvironment env, Map<String, DataStream<String>> streams) throws SQLException, ClassNotFoundException {

        //1、将订单表和订单详情表进行interval join（两个事实表）
        SingleOutputStreamOperator<OrderWide> orderWideStream = StreamJoin(streams);

        //2、与维度表进行join，通过查询phoenix进行join
        JoinDim(orderWideStream);
    }

    /**
     * 与维度表进行关联，增加redis缓存
     * @param orderWideStream
     */
    private void JoinDim(SingleOutputStreamOperator<OrderWide> orderWideStream) throws SQLException, ClassNotFoundException {

        orderWideStream.map(new RichMapFunction<OrderWide, OrderWide>() {

            private Jedis redis;
            private Connection connect;

            @Override
            public void open(Configuration parameters) throws Exception {
                connect = MyJDBCUtil.getConnect(Constant.PHOENIX_URL, Constant.PHOENIX_DRIVER);
                redis = MyJedisUtil.getRedis();
            }

            @Override
            public OrderWide map(OrderWide orderWide) throws Exception {
                //关联user_info表
                JSONObject userInfo = DimQuery
                        .getDimQuery(connect,redis,"DIM_USER_INFO", orderWide.getUser_id());
                //在phoenix中，查询出的字段名是大写的
                orderWide.setUser_gender(userInfo.getString("GENDER"));
                orderWide.calcAge(userInfo.getString("BIRTHDAY"));

                //关联dim_base_province表
                JSONObject baseProvince = DimQuery
                        .getDimQuery(connect, redis,"DIM_BASE_PROVINCE", orderWide.getProvince_id());
                orderWide.setProvince_3166_2_code(baseProvince.getString("ISO_3166_2"));
                orderWide.setProvince_area_code(baseProvince.getString("AREA_CODE"));
                orderWide.setProvince_name(baseProvince.getString("NAME"));
                orderWide.setProvince_iso_code(baseProvince.getString("ISO_CODE"));

                //关联sku_info表
                JSONObject skuInfo = DimQuery
                        .getDimQuery(connect, redis,"DIM_SKU_INFO", orderWide.getSku_id());
                orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
                orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                orderWide.setTm_id(skuInfo.getLong("TM_ID"));

                //关联spu_info表
                JSONObject spuInfo = DimQuery
                        .getDimQuery(connect, redis,"DIM_SPU_INFO", orderWide.getSpu_id());
                orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));

                //关联dim_base_category3表
                JSONObject category3 = DimQuery
                        .getDimQuery(connect, redis,"DIM_BASE_CATEGORY3", orderWide.getCategory3_id());
                orderWide.setCategory3_name(category3.getString("NAME"));

                //关联DIM_BASE_TRADEMARK表
                JSONObject baseTrademark = DimQuery
                        .getDimQuery(connect, redis,"DIM_BASE_TRADEMARK", orderWide.getTm_id());
                orderWide.setTm_name(baseTrademark.getString("TM_NAME"));

                return orderWide;
            }

            @Override
            public void close() throws Exception {

                if (connect != null) {
                    connect.close();
                }

                if (redis != null) {
                    redis.close();
                }
            }
        });


    }

    /**
     * 将订单表和订单详情表进行join，得到宽表
     *
     * 使用interval join的方式进行join，
     *      --1、需要使用事件时间
     *      --2、必须是keyby之后的数据
     */
    private SingleOutputStreamOperator<OrderWide> StreamJoin(Map<String, DataStream<String>> streams) {

        //获取数据，封装为对象，添加水印，按照订单id进行keyby

        KeyedStream<OrderInfo, Long> orderInfoStream = streams
                .get("dwd_order_info")
                .map(s -> JSON.parseObject(s, OrderInfo.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((ele, ts) -> ele.getCreate_ts())
                )
                .keyBy(OrderInfo::getId);


        KeyedStream<OrderDetail, Long> orderDetailStream = streams
                .get("dwd_order_detail")
                .map(s -> JSON.parseObject(s, OrderDetail.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((ele, ts) -> ele.getCreate_ts())
                )
                .keyBy(OrderDetail::getOrder_id);

        SingleOutputStreamOperator<OrderWide> result = orderInfoStream
                .intervalJoin(orderDetailStream)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo,
                                               OrderDetail orderDetail,
                                               Context ctx,
                                               Collector<OrderWide> out) throws Exception {

                        //将两个流join的结果作为OrderWide输出
                        out.collect(new OrderWide(orderInfo, orderDetail));
                    }
                });

        return result;
    }
}

